#include <arpa/inet.h>
#include <sys/socket.h>
#include <stdio.h>
#include <pthread.h>
#include <stdlib.h>
#include <netinet/in.h>
#include <unistd.h>
#include <sys/time.h>
#include <queue>
#include <iostream>

#include <boost/thread/exceptions.hpp>
#include <boost/thread/thread.hpp>
#include <boost/thread/mutex.hpp>

#define MAXPENDING 5
#define RCVBUFSIZE 512
#define MESSAGESIZE 51200
#define TIMEOUT 2

using std::queue;
using std::cout;
using std::endl;

struct Task {
    int sock;
   // char message[RCVBUFSIZE];
};

//boost::mutex queueMutex;
queue<Task> taskQueue;
pthread_mutex_t mutex = PTHREAD_MUTEX_INITIALIZER;

int workingThreads = 0;
pthread_mutex_t countMutex = PTHREAD_MUTEX_INITIALIZER;

int countTasks = 0;
//#include <sys/types.h>
//#include <netdb.h>
//#include <signal.h>
//#include <fcntl.h>
//#include <errno.h>


 int createTCPServerSocket(unsigned short port)
{
	int sock;                       
	struct sockaddr_in servAddr; 

	if ((sock = socket(PF_INET, SOCK_STREAM, IPPROTO_TCP)) < 0) 
	{
		cout << "create socket failed" << endl;	    
		return -1;
	}
      
	servAddr.sin_family = AF_INET;                
	servAddr.sin_addr.s_addr = htonl(INADDR_ANY); 
	servAddr.sin_port = htons(port);            

	if (bind(sock, (struct sockaddr *) &servAddr, sizeof(servAddr)) < 0)
	{
		cout << "bind failed" <<endl;	    
		return -1;
	}
	if (listen(sock, MAXPENDING) < 0)
	{
		cout << "listen failed" << endl;	    
		return -1;
	}
	return sock;
}

int acceptTCPConnection(int servSock)
{
	int clntSock;    
	struct sockaddr_in clntAddr; 	
	unsigned int clntLen;            
	clntLen = sizeof(clntAddr);
	if ((clntSock = accept(servSock, (struct sockaddr *) &clntAddr, &clntLen)) < 0)
	{
		cout << "accept failed" << endl;	    
		return -1;
	}
	//cout << "Handling client " <<  inet_ntoa(clntAddr.sin_addr) << endl;
	return clntSock;
}

void * workerThread(void *threadArgs)
{
	//pthread_detach(pthread_self()); 
    
    Task t;
    for (;;) 
    {
        //boost::mutex::scoped_lock lock(queueMutex);
        //queueMutex.lock();
        pthread_mutex_lock( &mutex );
        if (taskQueue.empty()) 
        {
            pthread_mutex_unlock( &mutex );
            //lock.unlock();
            sleep(2);
        } else {
            t = taskQueue.front();
            taskQueue.pop();
            //printf("%d\n", taskQueue.size());
            pthread_mutex_unlock( &mutex );
	    pthread_mutex_lock(&countMutex);
	    workingThreads++;  
	    struct timeval timeval;
	    gettimeofday( &timeval, NULL );
	    //timeinfo = localtime ( &rawtime );

	    if (workingThreads == 1) {
		    cout << "start work;" << (timeval.tv_sec % 1000)*1000000 + timeval.tv_usec << endl;
	    }
	    pthread_mutex_unlock(&countMutex);
            //lock.unlock();
            time_t seconds = time(NULL);
            time_t startTime = seconds;
            //printf("start %ld\n", seconds);
            //cout << "start clocks;" <<  clock() << endl;
            while (seconds < startTime + TIMEOUT) 
            {
                int * arr = new int[1000];
                for (int i = 0; i < 1000; i++) 
                {
                    arr[i] = i*i;
                }
                delete[] arr;
                seconds = time(NULL);	
            }
            //printf("end %ld\n", seconds);
            //cout << "end clocks;" <<  clock() << endl;
            int clntSock = t.sock;
            
            char response[6] = "Done!";
            if (send(clntSock, response, 6, 0) != 6) 
            {
                cout << "send failed" << endl;      
                return NULL;
            }
	    pthread_mutex_lock(&countMutex);
	    workingThreads--;
	    if (workingThreads == 0) {
		    gettimeofday( &timeval, NULL );
		    cout << "end work;" << (timeval.tv_sec % 1000)*1000000 + timeval.tv_usec << endl;
	    }
	    pthread_mutex_unlock(&countMutex);
        }      
    }
	return (NULL);
}

int handleTCPClient(int clntSocket)
{
	
	char buffer[RCVBUFSIZE];
	int recvMsgSize;                    

	if ((recvMsgSize = recv(clntSocket, buffer, RCVBUFSIZE, 0)) < 0) 
	{
		cout << "recv failed" << endl;	    
		return -1;
	}

    int totalSize = recvMsgSize;
	pthread_t threadID;              
	while (recvMsgSize > 0)     
	{		
        if (totalSize >= MESSAGESIZE) 
        {
            Task t;
            t.sock = clntSocket;             
            //boost::mutex::scoped_lock lock(queueMutex); 
            pthread_mutex_lock( &mutex );           
            taskQueue.push(t);
	    countTasks++;
            //printf("%d\n", totalSize);
            pthread_mutex_unlock( &mutex );
            totalSize = 0;
        }
		/*if (pthread_create(&threadID, NULL, workerThread, NULL) != 0)
		{
			printf("worker thread create failed");
			return -1;
		}
		printf("with thread %ld\n", (long int) threadID);*/
		if ((recvMsgSize = recv(clntSocket, buffer, RCVBUFSIZE, 0)) < 0)
		{
			cout << "recv failed" << endl;
			return -1;
		}
        totalSize += recvMsgSize;
	}

	cout << "finish client processing" << endl;
	//close(clntSocket);   
}

void * clientHandlerThread(void *threadArgs)
{
        int clntSock; 
        pthread_detach(pthread_self()); 
        clntSock = *(int *) threadArgs;

        handleTCPClient(clntSock);

        return (NULL);
}


int main(int argc, char ** argv) 
{    
	int servSock;
	int clntSock;     
	unsigned short servPort; 
	pthread_t threadID;    

	if (argc != 3)     
	{
		cout << "Usage: " <<  argv[0] << " <SERVER PORT> <number of clients>" << endl;
		return -1;
	}    
	
	int clientNumber = atoi(argv[2]);


    for (int i = 0; i < 10; ++i) {              
        if (pthread_create(&threadID, NULL, workerThread, NULL) != 0)
        {
            cout << "worker thread create failed" << endl;
            return -1;
        }
    }
    
	servPort = atoi(argv[1]);
	servSock = createTCPServerSocket(servPort);

    for (int i = 0; i < clientNumber; ++i) 
    {
            clntSock = acceptTCPConnection(servSock);               
            if (pthread_create(&threadID, NULL, clientHandlerThread, (void *) (&clntSock)) != 0)
            {
                    cout << "client handler thread create failed" << endl;
                    return -1;
            }
    }
    while (countTasks < 10*clientNumber) {
	sleep(1);
    }
    cout << countTasks << endl;
    sleep(3);
	return 0;
}